package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.dao.TagInfoDao;
import com.atguigu.userprofile.util.MyPropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskMergeSql {
    public static void main(String[] args) {
        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String upDBName = properties.getProperty("user-profile.dbname");

        SparkConf sparkConf = new SparkConf().setAppName("TaskMergeSql");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();


        sparkSession.sql("use user_profile220718");




        //1.获取业务日期
        String busiDate = args[1];


        //2.查询所有已启用标签的数据
        List<TagInfo> tagInfoWithOn = TagInfoDao.getTagInfoWithOn();
//        System.out.println(tagInfoWithOn);

        //3.动态拼接建表语句
        /**
         * create table if not exists up_tag_merge_20200614
         *   (uid String,
         *     tg_person_base_gender string,
         *     tg_person_base_agegroup string)
         *   ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
         *   location 'hdfs://hadoop102:8020/user_profile/user_profile/up_tag_merge_20200614'
         */

        String tableName = "up_tag_merge_" + busiDate.replace("-", "");
//        System.out.println(tableName);

        //获取建表语句中标签字段
        List<String> tagCodeWitOnList = tagInfoWithOn.stream().map(tagInfo -> tagInfo.getTagCode().toLowerCase() + " string").collect(Collectors.toList());

        String tagCodeSQL = StringUtils.join(tagCodeWitOnList, ",");



        String createSQL = "create table if not exists "+tableName+" \n" +
                "           (uid String,\n" +
                "             "+tagCodeSQL+") \n" +
                "           ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t' \n" +
                "           location '"+hdfsPath+"/"+upDBName+"/"+tableName+"'";

//        System.out.println(createSQL);

        sparkSession.sql(createSQL);

        //4.动态拼接查询语句
        /**
         *   c
         */

        //4.1动态拼接子查询
        List<String> subSelectSQL = tagInfoWithOn.stream().map(tagInfo -> "select uid,cast(tag_value as string) as tag_value,'" + tagInfo.getTagCode().toLowerCase() + "' as tag_code from " + tagInfo.getTagCode().toLowerCase() + " where dt = '" + busiDate + "'").collect(Collectors.toList());

        String unionSQL = StringUtils.join(subSelectSQL, " union all ");

        //4.2拼接查询语句
        List<String> pivotFiledList = tagInfoWithOn.stream().map(tagInfo -> "'" + tagInfo.getTagCode().toLowerCase() + "'").collect(Collectors.toList());
        String pivotFiled = StringUtils.join(pivotFiledList, ",");

        String selectSQL = " select * from (" + unionSQL + ") tg\n" +
                "           pivot (max(tag_value) as tag_value for tag_code in(" +
                ""+pivotFiled+"))";

        System.out.println(selectSQL);

        //5.动态拼接插入语句
        String insertSQL = "insert overwrite table " + tableName + " " + selectSQL;

        //6.执行SQL
        sparkSession.sql(insertSQL);
    }
}
